﻿using Microsoft.Extensions.DependencyInjection;
using Newtonsoft.Json;
using NLog;
using Stee.CAP8.MessageBus.Interface.Enums;
using Stee.CAP8.MessageBus.Interface.Interfaces;
using Stee.CAP8.MessageBus.Interface.Models;
using Stee.CAP8.MessageBus.Kafka.Configurations;
using Stee.CAP8.MessageBus.Kafka.Publisher;
using Stee.CAP8.MessageBus.Kafka.Subscriber;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Tasks;
using test.kafka;
using Test.EventBus2.Core.Serializaion;
using Test.EventBus2.Core.SubscribeStores;
using Test.EventBus2.Message.Events;

namespace Test.EventBus2.Core
{
    public class EventBus : IBus
    {

        private static readonly Logger Logger = LogManager.GetCurrentClassLogger();
        #region Topic Information

        private static readonly string topic = "first_topic";
        private static readonly string key = "first_key";
        private static readonly string groupID = "first_group";


        private static readonly TopicName topicNameObject = new TopicName(topic, key, MessageType.Event);

        #endregion Topic Information

        #region Publisher & Subscriber

        private static ISubscriber<TransportMessage> objectSubscriber;
        private static CancellationTokenSource objectSubscribeToken;

        private static IPublisher<TransportMessage> objectPublisher;
        private static CancellationTokenSource objectPublishToken;

        #endregion Publisher & Subscriber

        #region Service Provider & Others

        private static ServiceProvider serviceProvider;

        #endregion Service Provider & Others


        private ISubscribeStore _subscribeStore;
        private SimpleAssemblyQualifiedMessageTypeNameConvention messageTypeNameConvention;

        public EventBus(ISubscribeStore subscribeStore)
        {
            _subscribeStore = subscribeStore;
            messageTypeNameConvention = new SimpleAssemblyQualifiedMessageTypeNameConvention();
            objectPublisher = new PublisherImpl<TransportMessage>();
        }




        public  void HandleEvent()
        {

            objectSubscriber = new SubscriberImpl<TransportMessage>();

            objectSubscriber.Initialize(topicNameObject, (topicName, message, brokerMessage) =>
            {
#if Debug
                Console.WriteLine($" 收到消息topicName => {topicName}");
                Console.WriteLine($" message => {JsonConvert.SerializeObject(message)}");
                Console.WriteLine($" brokerMessage => {brokerMessage}");
#endif
                var typeObject = messageTypeNameConvention.GetType(message.Action);
                var handlers = _subscribeStore.GetMessageHandlersForMessage(typeObject).Cast<IHandleMessages>();
                var messageObject = JsonConvert.DeserializeObject(message.Body, typeObject);
                var func = _subscribeStore.GetMessageHandlerFunction(typeObject);
                func?.DynamicInvoke(messageObject);

                return objectSubscriber.Commit(brokerMessage);
            }, OnSubscriberError);

            if (objectSubscriber.IsInitialized)
            {
                objectSubscriber.Subscribe();
            }


        }


        public void PublishMessage<TMessage>(TMessage @event)

        {

            var typeName = Shims.GetSimpleAssemblyQualifiedName(@event.GetType());

            var transportmessage = new TransportMessage
            {
                Action = typeName,
                Body = JsonConvert.SerializeObject(@event),
                CreateTime = DateTime.Now,
                Tag = "saga"
            };

            objectPublisher.Initialize(topicNameObject, response => response != null, OnPublisherError);
            if (objectPublisher.IsInitialized)
            {
                objectPublishToken = new CancellationTokenSource();
                if (objectPublisher.Publish(transportmessage, objectPublishToken.Token))
                {

                    //Console.WriteLine($"Publish Person object => {transportmessage}");
                }
                else
                {
                    //Console.WriteLine($"[Error] Publish Person object failed => {transportmessage}");
                }
            }
           
        }


        /// <summary>
        /// Mock example method name format, "On<Action><Response>".
        /// Error handler for subscriber.
        /// </summary>
        /// <param name="errorCode">Broker error code</param>
        /// <param name="isFatal">Flag indicating fatal error</param>
        private static void OnSubscriberError(string errorCode, bool isFatal)
        {
            if (string.IsNullOrEmpty(errorCode))
            {
                Logger.Warn($"Unexpected subscriber error, no error code returned. Is fatal: {isFatal}");
            }
            else
            {
                Logger.Error($"Subscriber error: {errorCode}, Is fatal: {isFatal}");
            }
        }

        /// <summary>
        /// Mock example method name format, "On<Action><Response>".
        /// Publish error handler for broker publish error.
        /// </summary>
        /// <param name="errorCode">Publish error code</param>
        /// <param name="isFatal">Flag indicating fatal error</param>
        private static void OnPublisherError(string errorCode, bool isFatal)
        {
            if (string.IsNullOrEmpty(errorCode))
            {
                Logger.Warn($"Unexpected publisher error, no error code returned. Is fatal: {isFatal}");
            }
            else
            {
                Logger.Error($"Publisher error: {errorCode}, Is fatal: {isFatal}");
            }
        }




    }


}
